/*
主题的发布订阅客户端实现
    1. 为用户提供创建主题等一系列操作的接口
    2. 为dispatcher模块提供服务端发送消息推送请求的路由接口
*/

#include "../Common/Net.hpp"
#include "Requestor.hpp"

namespace RpcModule
{
    namespace ClientModule
    {
        class TopicManager
        {
        public:
            using ptr = std::shared_ptr<TopicManager>;
            // key -- 主题名称    ||    message -- 要发送的消息主体
            using TopicCallBack = std::function<void(const std::string& key, const std::string& message)>;
            TopicManager(const Requestor::ptr& requestor) :_requestor(requestor) {}
            // 创建主题 -- 需要指出创建的主题名称
            bool Create(const BaseConnection::ptr& conn, const std::string& key) 
            {
                return CommonRequest(conn, key, TopicOptype::TOPIC_CREATE);
            }
            // 删除主题 -- 需要指出删除的主题名称
            bool Remove(const BaseConnection::ptr& conn, const std::string& key) 
            {
                return CommonRequest(conn, key, TopicOptype::TOPIC_REMOVE);
            }
            // 主题的订阅 -- 订阅的主题名称，当收到该主题消息时的回调函数
            bool SubScribe(const BaseConnection::ptr& conn, const std::string& key, const TopicCallBack& cb) 
            {
                // 主题的订阅有所不同，因为需要维护主题与主题消息到来时的回调函数的关系
                // 也就是需要维护unordered_map<std::string, TopicCallBack> _topic_callbacks;
                bool ret = CommonRequest(conn, key, TopicOptype::TOPIC_SUBSCRIBE);
                if(ret == false)
                {
                    LOGERROR("主题的订阅操作失败");
                    return false;
                }
                AddCallBack(key, cb);
                return true;
            }
            // 主题的取消订阅 -- 取消订阅的主题名称
            bool UnSubScribe(const BaseConnection::ptr& conn, const std::string& key) 
            {
                bool ret = CommonRequest(conn, key, TopicOptype::TOPIC_CANCEL);
                if(ret == false)
                {
                    LOGERROR("主题的取消订阅操作失败");
                    return false;
                }
                DeleteCallBack(key);    
                return true;
            }
            // 主题消息的推送 -- 要向什么主题，推送什么消息
            bool Publish(const BaseConnection::ptr& conn, const std::string& key, const std::string& message) 
            {
                return CommonRequest(conn, key, TopicOptype::TOPIC_PUBLISH, message);
            }
            // 为dispatcher模块提供的，当收到服务端的消息推送请求时的回调函数
            void OnPublish(const BaseConnection::ptr& conn, const TopicRequest::ptr& message) 
            {
                // 1. 收到消息推送请求，先解析请求
                TopicOptype optype = message->GetTopicOptype();
                if(optype != TopicOptype::TOPIC_PUBLISH)
                {
                    LOGERROR("收到了错误类型的消息，应该为TOPIC_PUBLISH");
                    return;
                }
                std::string topic_name = message->GetTopicKey();
                std::string response_message = message->GetTopicMessage();
                // 2. 调用注册的回调函数
                TopicCallBack callback = GetCallBack(topic_name);
                if(!callback)
                {
                    LOGERROR("不存在%s消息的回调函数，无法处理该类型消息", topic_name.c_str());
                    return;
                }
                return callback(topic_name, response_message);
            }
        private:
            // 组织请求，向服务端发起请求，并接收结果，判断结果是否正确
            bool CommonRequest(const BaseConnection::ptr& conn, const std::string& key,
                TopicOptype optype, const std::string& message = "")
            {
                // 1. 发起创建主题的请求
                TopicRequest::ptr topic_request = MessageFactory::CreateMessage<TopicRequest>();
                topic_request->SetId(UUID::GetUuid());
                topic_request->SetMType(MType::REQ_TOPIC);
                topic_request->SetTopicKey(key);
                topic_request->SetTopicOptype(optype);
                if(optype == TopicOptype::TOPIC_PUBLISH)
                    topic_request->SetTopicMessage(message);
                BaseMessage::ptr response;
                bool ret = _requestor->Send(conn, topic_request, response);
                // 2. 接收服务端的响应
                if(ret == false)
                {
                    LOGERROR("客户端向主题中心发起%s请求，接收到的响应失败", key.c_str());
                    return false;
                }
                TopicResponse::ptr topic_response = std::dynamic_pointer_cast<TopicResponse>(response);
                if(!topic_response)
                {
                    LOGERROR("向下转化失败");
                    return false;
                }
                RCode rcode = topic_response->GetRcode();
                if(rcode != RCode::RCODE_OK)
                {
                    LOGERROR("客户端接收到响应，但是发起的请求%s处理失败%s", key.c_str(), ErrReason(rcode).c_str());
                    return false;
                }
                // 3. 走到这里，说明发起请求成功，接收到的响应也是成功处理的
                return true;
            }
            void AddCallBack(const std::string& key, const TopicCallBack& cb)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callbacks.insert(std::make_pair(key, cb));
            }
            void DeleteCallBack(const std::string& key)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callbacks.erase(key);
            }
            TopicCallBack GetCallBack(const std::string& key)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _topic_callbacks.find(key);
                if(it == _topic_callbacks.end()) return TopicCallBack();
                return it->second;
            }
        private:
            std::mutex _mutex;
            // 接收到特定主题消息请求的回调函数组织
            std::unordered_map<std::string, TopicCallBack> _topic_callbacks;
            // 使用requestor进行请求的发送
            Requestor::ptr _requestor;
        };
    }
}